package com.ruoyi.system.rocketmq.netty;

import com.alibaba.fastjson.JSONObject;
import com.ruoyi.common.domain.constant.RedisKey;
import com.ruoyi.common.domain.constant.RoleConstant;
import com.ruoyi.common.redis.service.RedisService;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;

/**
 * Netty WebSocket 长连接服务端（含心跳机制和 Redis 清理）
 */
@Slf4j
@Component
public class NettyWebSocketServer {

    @Value("${netty.websocket.port:19204}")
    private int port;

    @Resource
    private RedisService redisService;

    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    private Channel serverChannel;

    /** 全部客户端连接 */
    private static final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    /** channel 属性：userId */
    private static final AttributeKey<String> USER_ID = AttributeKey.valueOf("userId");

    @PostConstruct
    public void start() {
        log.info("启动Netty服务 port:{}",port);
        bossGroup = new NioEventLoopGroup(1);
        workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<Channel>() {
                        @Override
                        protected void initChannel(Channel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new HttpServerCodec());
                            pipeline.addLast(new HttpObjectAggregator(65536));
                            pipeline.addLast(new ChunkedWriteHandler());
                            pipeline.addLast(new WebSocketServerProtocolHandler("/ws", null, true));

                            // 心跳机制：读超时 60s，写空闲 30s
                            pipeline.addLast(new IdleStateHandler(60, 30, 0));

                            pipeline.addLast(new SimpleChannelInboundHandler<TextWebSocketFrame>() {

                                @Override
                                public void channelActive(ChannelHandlerContext ctx) {
                                    channels.add(ctx.channel());
                                    log.info("新客户端连接: {}", ctx.channel().id());
                                }

                                @Override
                                public void channelInactive(ChannelHandlerContext ctx) {
                                    channels.remove(ctx.channel());

                                    String userId = ctx.channel().attr(USER_ID).get();
                                    
                                    if (userId != null) {
                                        String redisKey = String.format(RedisKey.ONLINE_AUDITORS, RoleConstant.AUDITORS);
                                        redisService.removeZSet(redisKey, userId);
                                        log.info("客户端断开: {}, 移除在线用户 {}", ctx.channel().id(), userId);
                                    } else {
                                        log.info("客户端断开: {}, 未绑定 userId", ctx.channel().id());
                                    }
                                }

                                @Override
                                protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) {
                                    String text = msg.text();

                                    // 心跳响应
                                    if ("pong".equalsIgnoreCase(text)) {
                                        log.debug("收到客户端心跳响应: {}", ctx.channel().id());
                                        return;
                                    }

                                    JSONObject json = JSONObject.parseObject(text);
                                    String userId = String.valueOf(json.get("userId"));


                                    // 模拟客户端发送 userId
                                    ctx.channel().attr(USER_ID).set(userId);
                                    String redisKey = String.format(RedisKey.ONLINE_AUDITORS, RoleConstant.AUDITORS);
                                    redisService.zAdd(redisKey, userId, System.currentTimeMillis());
                                    log.info("收到消息 [{}]: userId={}", ctx.channel().id(), userId);
//                                    ctx.channel().writeAndFlush(new TextWebSocketFrame("服务端收到: " + userId));
                                }

                                @Override
                                public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                                    if (evt instanceof IdleStateEvent e) {
                                        if (e.state() == IdleState.READER_IDLE) {
                                            // 客户端长时间未回应，清理 Redis 并关闭连接
                                            String userId = ctx.channel().attr(USER_ID).get();
                                            if (userId != null) {
                                                String redisKey = String.format(RedisKey.ONLINE_AUDITORS, RoleConstant.AUDITORS);
                                                redisService.removeZSet(redisKey, userId);
                                                log.warn("客户端长时间未响应，清理 Redis 并关闭连接: {}, userId={}",
                                                        ctx.channel().id(), userId);
                                            } else {
                                                log.warn("客户端长时间未响应，关闭连接: {}, 未绑定 userId", ctx.channel().id());
                                            }
                                            ctx.close();
                                        } else if (e.state() == IdleState.WRITER_IDLE) {
                                            // 写空闲时发送心跳
                                            ctx.channel().writeAndFlush(new TextWebSocketFrame("ping"));
                                        }
                                    } else {
                                        super.userEventTriggered(ctx, evt);
                                    }
                                }

                                @Override
                                public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                                    log.error("异常", cause);
                                    ctx.close();
                                }
                            });
                        }
                    })
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            ChannelFuture future = bootstrap.bind(port).sync();
            serverChannel = future.channel();
            log.info("🚀 Netty WebSocket 服务启动成功，端口: {}", port);
        } catch (Exception e) {
            log.error("❌ Netty WebSocket 启动失败", e);
        }
    }

    @PreDestroy
    public void stop() {
        try {
            if (serverChannel != null) serverChannel.close();
            if (bossGroup != null) bossGroup.shutdownGracefully();
            if (workerGroup != null) workerGroup.shutdownGracefully();
            log.info("🛑 Netty WebSocket 服务关闭成功");
        } catch (Exception e) {
            log.error("关闭 Netty WebSocket 出错", e);
        }
    }

    /** 给所有连接的客户端发消息 */
    public void broadcast(String msg) {
        channels.writeAndFlush(new TextWebSocketFrame(msg));
    }

    /** 给指定 userId 发消息 */
    public void sendToUser(String userId, String msg) {
        channels.forEach(ch -> {
            String uid = ch.attr(USER_ID).get();
            if (userId.equals(uid)) {
                ch.writeAndFlush(new TextWebSocketFrame(msg));
            }
        });
    }

    /** 获取当前所有连接 */
    public ChannelGroup getChannels() {
        return channels;
    }
}
